package com.lmq

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.functions.{col, udf}
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, SparkSession}
/**
 * 第2步 预处理之过滤:
 * 删除在整个数据集中出现次数少于5次的ItemId，删除只包含一个ItemId的会话
 *
 */
object filterLen {
  Logger.getLogger("org.apache.spark")
    .setLevel(Level.WARN)

  val spark = SparkSession.builder()
    .master("local[*]")
    .appName("Test")
    .getOrCreate()

//  def main(args: Array[String]): Unit = {
//
//    Logger.getLogger("org.apache.spark")
//      .setLevel(Level.WARN)
//
//    val spark = SparkSession.builder()
//      .master("local[*]")
//      .appName("Test")
//      .getOrCreate()


    def main(args: Array[String]): Unit = {
      val utils = new utils()

      val schema = StructType(
        Array(
          StructField("SessionId", IntegerType, nullable = true),
          StructField("TimeStr", StringType, nullable = true),
          StructField("ItemId", IntegerType, nullable = true),
          StructField("Context", StringType, nullable = true),
          StructField("Time",DoubleType, nullable = false),
          StructField("TimeRank",IntegerType, nullable = true)
        )
      )
      val v: DataFrame = spark.read
        .schema(schema)
        .option("header",value = false)
        .csv("file:///home/iptv/yoochoose/Reslt.csv/1over64.csv")
//        .csv("D:\\pythonProject\\pythonProject\\ComparisonWithYoochoose\\src\\yoochoose-clicks.csv")
//      df.show(false)

//      val timeTransfer = udf((x:String)=>  utils.timeStr2Tsp(x) )
//      df.select(
//        col("SessionId"),
//        col("TimeStr"),
//        col("ItemId"),
//        col("Context"),
//        timeTransfer( col("TimeStr") ).alias("Time")
//      )
//        .createOrReplaceTempView("allData")
//
//      val v: DataFrame = spark.sql(""" select * ,rank() over (partition by SessionId order by Time) as
//    `TimeRank` from allData""")
      v.printSchema()
      println("=====================================")
      v.show(false)
//      println(v.count())
      v.createTempView("v")
//
//      spark.sql(
//        """
//          |
//          | select * from
//          |( select ItemId, count(1) as cnt
//          |from v
//          |group by ItemId) A where cnt>=5
//          |""".stripMargin)
//        //        .count())
//        .show(1000,truncate = false)
//      println(
        spark.sql(
        """select v.* from
          |(select ItemId from
          |( select ItemId, count(1) as cnt
          |from v
          |group by ItemId) A where cnt>=5) B
          |left join v
          |on B.ItemId=v.ItemId
          |""".stripMargin)
//        .count())
//        .show(1000,truncate = false)

//      val xtable =
        spark.sql(
        """select w.SessionId,w.ItemId,w.Context,w.Time,
          |rank() over (partition by SessionId order by Time) as TimeRank
          |from
          |(select v.* from
          |(select ItemId from
          |( select ItemId, count(1) as cnt
          |from v
          |group by ItemId) A where cnt>=5) B
          |left join v
          |on B.ItemId=v.ItemId) w
          |""".stripMargin).createTempView("xtable")
//        .show(false)
      spark.sql(
        """
          |select xtable.SessionId,xtable.ItemId,xtable.Context,xtable.Time
          |from
          |(select SessionId from
          |(select SessionId,sum(1) as cnts from xtable group by SessionId) groupedtable where cnts>4
          |) ltable
          |left join xtable
          |on ltable.SessionId=xtable.SessionId
          |""".stripMargin)
        .createTempView("unsortTable")
        spark.sql(
          """
            |select * from unsortTable
            |order by SessionId,Time
            |""".stripMargin)
        .coalesce(1).write
        .option("header","false")
          // TODO: here is the storage location, note that this directory is renamed to OneOsixfour_V3
          //      in the later processing or or is copied to a new directory with  name OneOsixfour_V3 .But you could rename the OneOsixfour_V3 to OneOsixfour_v2 in other
          //      code file if you like . The reason why I will rename directory is that I am testing my code ,and make sure
          //      data is correctly processed.
        .csv("file:///home/iptv/yoochoose/OneOsixfour_v2")
      println("Done.")
//        .show(1000,false)


      // remove sessions which the length is less than 1 and apearance is less than 5.

      //    v.coalesce(1).write
      //      .option("header","true")
      //      .csv("file:///home/iptv/yoochoose/Reslt.csv")

      //      .toPandas()\
      //    .to_csv("helloResult.csv",index=None)







    }

  }
